package pubsub

import (
	"errors"
	"log/slog"
	"strings"
	"sync"

	"gitlab.local/DO-module/new-filemanage-module/src/types"
)

type Publisher interface {
	// Publish 发布事件
	Publish(types.Event)
	// 订阅者数量
	SubCount(topic string) int
	// 注册事件源
	Register(secretKey string, topics []types.TopicInfo, sessionId string) error
	// 注销事件源
	DeregisterPublisher(sessionId string)
}

type registry struct {
	topics    []types.TopicInfo
	sessionId string
}

type PubSub struct {
	logger            *slog.Logger
	subscriptions     map[string][]types.EventPublisher // key 是 topic，value 是订阅列表
	sLock             sync.RWMutex
	topicRegistry     map[string]*registry // key 是 SecretKey，value 是注册的 topics
	rLock             sync.RWMutex
	topicLastEventMap map[string]*types.Event
	mLock             sync.Mutex
}

func New(logger *slog.Logger) *PubSub {
	return &PubSub{
		logger:            logger.With("ctx", "pubsub"),
		subscriptions:     make(map[string][]types.EventPublisher),
		topicRegistry:     make(map[string]*registry),
		topicLastEventMap: make(map[string]*types.Event),
	}
}

func (s *PubSub) Publish(e types.Event) {
	s.sLock.RLock()
	defer s.sLock.RUnlock()
	go s.updateTopicLastEventMap(&e)
	s.logger.Debug("publish", "topic", e.Topic)
	if conns, ok := s.subscriptions[e.Topic]; ok {
		for _, c := range conns {
			if c != nil {
				err := c.Publish(e)
				if err != nil {
					s.logger.Error("send message error", "err", err)
				}
			}
		}
	}
}

func (s *PubSub) PublishCheck(secretKey, topic string) error {
	s.rLock.RLock()
	defer s.rLock.RUnlock()
	if r, ok := s.topicRegistry[secretKey]; ok {
		if !constinsTopic(r.topics, topic) {
			return types.ErrPublishShouldRegisterFirst
		}
	}
	return nil
}

func (s *PubSub) Register(secretKey string, topics []types.TopicInfo, sessionId string) error {
	s.rLock.Lock()
	defer s.rLock.Unlock()
	for _, v := range s.topicRegistry {
		for _, topic := range topics {
			if constinsTopic(v.topics, topic.Name) {
				return types.ErrRegisterForbidden
			}
		}
	}
	s.topicRegistry[secretKey] = &registry{
		topics:    topics,
		sessionId: sessionId,
	}
	s.logger.Debug("registered", "key", secretKey)
	return nil
}

func (s *PubSub) DeregisterPublisher(sessionId string) {
	s.rLock.Lock()
	defer s.rLock.Unlock()
	for k, v := range s.topicRegistry {
		if v.sessionId == sessionId {
			s.logger.Debug("deregistered", "key", k)
			delete(s.topicRegistry, k)
		}
	}
}

func (s *PubSub) Subcribe(topic string, publisher types.EventPublisher) {
	if len(topic) == 0 {
		s.logger.Warn("topic is empty. subcribe do nothing")
		return
	}
	topic, ok := strings.CutPrefix(topic, ":")
	s.sLock.Lock()
	defer s.sLock.Unlock()
	s.subscriptions[topic] = append(s.subscriptions[topic], publisher)
	if ok {
		if e, exists := s.topicLastEventMap[topic]; exists {
			publisher.Publish(*e)
		}
	}
	s.logger.Debug("subcribed", "topic", topic, "sessionId", publisher.SessionId())
}

func (s *PubSub) Unsubcribe(topic string, publisher types.EventPublisher) {
	if len(topic) == 0 {
		s.logger.Warn("topic is empty. subcribe do nothing")
		return
	}
	topic, _ = strings.CutPrefix(topic, ":")
	s.sLock.Lock()
	defer s.sLock.Unlock()
	if conns, ok := s.subscriptions[topic]; ok {
		subCount := len(conns)
		for i, c := range conns {
			if c == publisher {
				conns[i] = nil
			}
			if conns[i] == nil {
				subCount--
			}
		}
		if subCount <= 0 {
			delete(s.subscriptions, topic)
		}
	}
	s.logger.Debug("unsubcribed", "topic", topic, "sessionId", publisher.SessionId())
}

func (s *PubSub) Last(topic string, publisher types.EventPublisher) {
	if len(topic) == 0 {
		s.logger.Warn("topic is empty. subcribe do nothing")
		return
	}
	topic, _ = strings.CutPrefix(topic, ":")
	if e, exists := s.topicLastEventMap[topic]; exists {
		publisher.Publish(*e)
		return
	}
	publisher.Publish(types.Error{Err: errors.New("not found"), Code: 40400}.ToEvent())
}

func (s *PubSub) UnsubcribeAll(publisher types.EventPublisher) {
	s.sLock.Lock()
	defer s.sLock.Unlock()
	for k, v := range s.subscriptions {
		subCount := len(v)
		for i, c := range v {
			if c == publisher {
				v[i] = nil
			}
			if v[i] == nil {
				subCount--
			}
		}
		if subCount <= 0 {
			delete(s.subscriptions, k)
		}
	}
}

func (s *PubSub) SubCount(topic string) int {
	s.sLock.RLock()
	defer s.sLock.RUnlock()
	count := 0
	for i := range s.subscriptions[topic] {
		if s.subscriptions[topic][i] != nil {
			count++
		}
	}
	return count
}

func (s *PubSub) ListTopics() []types.ExtendTopicInfo {
	s.rLock.RLock()
	defer s.rLock.RUnlock()
	var topics []types.ExtendTopicInfo
	for _, v := range s.topicRegistry {
		for _, topic := range v.topics {
			topics = append(topics, types.ExtendTopicInfo{TopicInfo: topic, SubscriberCount: s.SubCount(topic.Name)})
		}
	}
	return topics
}

func (s *PubSub) updateTopicLastEventMap(e *types.Event) {
	s.mLock.Lock()
	defer s.mLock.Unlock()
	s.topicLastEventMap[e.Topic] = e
}

func constinsTopic(topics []types.TopicInfo, topic string) bool {
	for _, t := range topics {
		if t.Name == topic {
			return true
		}
	}
	return false
}
